package com.hliushi.spark.rdd

import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/16 22:21
 */
class Broadcast {


  /**
   * 正常情况下使用Task拉取数据的时候, 会将数据拷贝到Executor中多次
   * 但是使用广播变量的时候只会复制一份数据到Executor中
   * .    所以在两种情况下特别适合使用广播变量
   * .        一个Executor中有多个Task的时候
   * .        一个变量比较大的时候
   * .
   * 而在Spark中还有一个约定俗成的做法, 当一个RDD很大并且还需要和另一个RDD执行join的时候
   * 可以将较小的RDD广播出去, 然后使用RDD的算子map中直接join, 从而实现在Map端join
   * #    这里和MapReduce中的Map_Join类似, mr做法是将小表放在分布式缓存中
   */

  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("source_analysis")

  val sc = new SparkContext(conf)


  /**
   * 资源占用比较大, 有是个对应的value
   */
  @Test
  def bc1(): Unit = {
    // 数据, 假装这个数据很大, 大概有100M
    val map = Map("Spark" -> "http://spark.apache.cn", "Scala" -> "http://www.scala-lang.org")

    // 将其中的Spark和Scala转为对应的网址
    val sourceRDD = sc.parallelize(Seq("Spark", "Scala"))
    val result = sourceRDD.map(item => map(item)).collect()

    result.foreach(println(_))
    sc.stop()
  }

  /**
   * 使用广播, 大幅度减少value的复制
   */
  @Test
  def bc2(): Unit = {
    // 数据, 假装这个数据很大, 大概有100M
    val map = Map("Spark" -> "http://spark.apache.cn", "Scala" -> "http://www.scala-lang.org")

    // 创建广播
    val bc = sc.broadcast(map)

    // 将其中的Spark和Scala转为对应的网址
    val sourceRDD = sc.parallelize(Seq("Spark", "Scala"))

    // 在算子中使用广播变量代替直接引用集合, 只会复制和executor一样的数量
    // 在使用广播之前, 复制了map 了task数量份
    // 使用广播以后, 复制次数和executor数量一致
    val result = sourceRDD.map((item: String) => bc.value(item)).collect()
    // val result2 = sourceRDD.map(bc.value).collect()

    result.foreach(println(_))
    sc.stop()
  }
}